package com.yyy.tcc.service.compensation;

import com.yyy.tcc.entity.CompensationRecord;
import com.yyy.tcc.entity.TransferRecord;
import com.yyy.tcc.service.AccountService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.*;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;

@Service
public class ScheduledCompensationService {
    private static final Logger log = LoggerFactory.getLogger(ScheduledCompensationService.class);
    private static final int MAX_RETRY_COUNT = 3;

    private final DataSource dataSource;
    private KafkaTemplate<String, String> kafkaTemplate;
    private AccountService accountService;

    @Autowired
    public ScheduledCompensationService(DataSource dataSource,
                                      KafkaTemplate<String, String> kafkaTemplate,
                                      AccountService accountService) {
        this.dataSource = dataSource;
        this.kafkaTemplate = kafkaTemplate;
        this.accountService = accountService;
    }

    /**
     * 用于测试的构造函数
     */
    public ScheduledCompensationService(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    /**
     * 为测试提供AccountService的setter方法
     */
    public void setAccountService(AccountService accountService) {
        this.accountService = accountService;
    }

    /**
     * 为测试提供KafkaTemplate的setter方法
     */
    public void setKafkaTemplate(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 每5分钟执行一次补偿
     */
    @Scheduled(fixedRate = 300000)
    public void compensateTimeoutTransactions() {
        log.info("开始执行超时事务补偿...");

        try (Connection conn = dataSource.getConnection()) {
            // 查询超时的事务记录
            List<TransferRecord> timeoutRecords = findTimeoutTransactions(conn);

            for (TransferRecord record : timeoutRecords) {
                try {
                    // 检查补偿记录状态
                    CompensationRecord compensationRecord = getCompensationRecord(record.getTransactionId());

                    if (compensationRecord == null) {
                        // 没有补偿记录，创建新的补偿记录并执行补偿
                        log.info("创建新的补偿记录: {}", record.getTransactionId());
                        createCompensationRecord(record.getTransactionId());
                        executeCompensation(record);
                    } else {
                        // 根据补偿记录状态处理
                        switch (compensationRecord.getStatus()) {
                            case "SUCCESS":
                                // 补偿成功，跳过
                                log.info("补偿已成功，跳过: {}", record.getTransactionId());
                                continue;
                            case "FAILED":
                                // 补偿失败，检查重试次数
                                if (compensationRecord.getRetryCount() < MAX_RETRY_COUNT) {
                                    // 未超过最大重试次数，重试补偿
                                    log.info("开始第{}次重试补偿: {}",
                                            compensationRecord.getRetryCount() + 1,
                                            record.getTransactionId());
                                    executeCompensation(record);
                                    updateCompensationRetryCount(compensationRecord);
                                } else {
                                    // 超过最大重试次数，发送告警
                                    log.error("补偿重试次数超过上限: {}, 最大重试次数: {}",
                                            record.getTransactionId(), MAX_RETRY_COUNT);
                                    sendAlertNotification(record);
                                }
                                break;
                            case "PENDING":
                                // 待处理状态，执行补偿
                                log.info("处理待执行的补偿: {}", record.getTransactionId());
                                executeCompensation(record);
                                break;
                            default:
                                log.warn("未知的补偿状态: {}, 事务ID: {}",
                                        compensationRecord.getStatus(), record.getTransactionId());
                        }
                    }
                } catch (Exception e) {
                    log.error("补偿执行失败: {}", record.getTransactionId(), e);
                    updateCompensationStatus(record.getTransactionId(), "FAILED");
                }
            }
        } catch (SQLException e) {
            log.error("补偿任务执行失败", e);
        }
    }

    /**
     * 查询超时的事务记录
     */
    private List<TransferRecord> findTimeoutTransactions(Connection conn) throws SQLException {
        List<TransferRecord> records = new ArrayList<>();
        String sql = "SELECT * FROM transfer_record WHERE status = 'TRYING' " +
                    "AND create_time < ? FOR UPDATE";

        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
            // 获取30分钟前的时间
            Timestamp timeout = Timestamp.valueOf(LocalDateTime.now().minusMinutes(30));
            stmt.setTimestamp(1, timeout);

            try (ResultSet rs = stmt.executeQuery()) {
                while (rs.next()) {
                    TransferRecord record = new TransferRecord();
                    record.setId(rs.getLong("id"));
                    record.setTransactionId(rs.getString("transaction_id"));
                    record.setFromUserId(rs.getString("from_user_id"));
                    record.setToUserId(rs.getString("to_user_id"));
                    record.setAmount(rs.getBigDecimal("amount"));
                    record.setStatus(rs.getString("status"));
                    records.add(record);
                }
            }
        }
        return records;
    }

    /**
     * 创建补偿记录
     */
    private void createCompensationRecord(String transactionId) throws SQLException {
        String sql = "INSERT INTO compensation_record (transaction_id, status, retry_count) " +
                    "VALUES (?, 'PENDING', 0)";

        try (Connection conn = dataSource.getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            stmt.setString(1, transactionId);
            stmt.executeUpdate();
        }
    }

    /**
     * 发送告警通知
     */
    private void sendAlertNotification(TransferRecord record) {
        // 这里实现告警通知逻辑，可以是发送邮件、短信等
        log.warn("发送告警通知: 事务{}补偿失败，已达到最大重试次数", record.getTransactionId());
    }

    /**
     * 执行补偿操作
     */
    private void executeCompensation(TransferRecord record) throws SQLException {
        log.info("执行补偿操作: {}", record.getTransactionId());
        try {
            accountService.cancel(record.getTransactionId());
            updateCompensationStatus(record.getTransactionId(), "SUCCESS");
        } catch (Exception e) {
            log.error("补偿执行失败: {}", record.getTransactionId(), e);
            updateCompensationStatus(record.getTransactionId(), "FAILED");
            throw e;
        }
    }

    /**
     * 更新补偿记录重试次数
     */
    private void updateCompensationRetryCount(CompensationRecord record) throws SQLException {
        String sql = "UPDATE compensation_record SET retry_count = retry_count + 1 " +
                    "WHERE transaction_id = ?";

        try (Connection conn = dataSource.getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            stmt.setString(1, record.getTransactionId());
            stmt.executeUpdate();
        }
    }

    /**
     * 获取补偿记录
     */
    private CompensationRecord getCompensationRecord(String transactionId) throws SQLException {
        String sql = "SELECT * FROM compensation_record WHERE transaction_id = ?";

        try (Connection conn = dataSource.getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            stmt.setString(1, transactionId);
            ResultSet rs = stmt.executeQuery();

            if (rs.next()) {
                CompensationRecord record = new CompensationRecord();
                record.setTransactionId(rs.getString("transaction_id"));
                record.setStatus(rs.getString("status"));
                record.setRetryCount(rs.getInt("retry_count"));
                return record;
            }
        }
        return null;
    }

    /**
     * 更新补偿状态
     */
    private void updateCompensationStatus(String transactionId, String status) throws SQLException {
        String sql = "UPDATE compensation_record SET status = ? WHERE transaction_id = ?";

        try (Connection conn = dataSource.getConnection();
             PreparedStatement stmt = conn.prepareStatement(sql)) {
            stmt.setString(1, status);
            stmt.setString(2, transactionId);
            stmt.executeUpdate();
        }
    }

    /**
     * 关闭定时任务服务
     */
    public void shutdown() {
        // 如果有需要清理的资源，在这里处理
        if (kafkaTemplate != null) {
            try {
                kafkaTemplate.destroy();
            } catch (Exception e) {
                log.error("关闭Kafka服务失败", e);
            }
        }
    }

}
